package cn.monkey.io.transport.core;

import io.netty.util.ReferenceCountUtil;
import org.reactivestreams.Subscription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.CoreSubscriber;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Operators;

import java.util.ArrayDeque;
import java.util.Queue;

public abstract class FluxReceive extends Flux<Object> implements Subscription, Disposable {
    private static final Logger log = LoggerFactory.getLogger(FluxReceive.class);
    protected boolean disposed;
    protected CoreSubscriber<? super Object> receiver;
    protected Queue<Object> receiverQueue;
    protected volatile Throwable error;
    protected long receiverDemand;

    protected static void clearQueue(Queue<Object> q) {
        if (q == null) {
            return;
        }
        if (q.isEmpty()) {
            return;
        }
        Object o;
        while ((o = q.poll()) != null) {
            ReferenceCountUtil.release(o);
        }
    }

    protected void terminateReceiver(Queue<Object> q, CoreSubscriber<? super Object> r) {
        if (q != null) {
            clearQueue(q);
        }
        Throwable ex = this.error;
        this.receiver = null;
        if (ex != null) {
            r.onError(ex);
        } else {
            r.onComplete();
        }
    }


    protected void drainReceiver() {
        for (; ; ) {
            final Queue<Object> q = this.receiverQueue;
            final CoreSubscriber<? super Object> r = this.receiver;
            if (r == null) {
                clearQueue(q);
                return;
            }
            if (this.disposed) {
                clearQueue(q);
                return;
            }
            long rd = this.receiverDemand;
            long offset = 0L;
            while (offset <= rd) {
                Object v = q == null ? null : q.poll();
                boolean empty = v == null;
                if (this.disposed && empty) {
                    this.terminateReceiver(q, r);
                    return;
                }
                if (empty) {
                    break;
                }
                try {
                    r.onNext(v);
                } finally {
                    try {
                        ReferenceCountUtil.release(v);
                    } catch (Throwable e) {
                        this.error = e;
                        clearQueue(q);
                        terminateReceiver(q, r);
                    }
                }
                offset++;
            }
        }
    }

    protected void startReceiver(CoreSubscriber<? super Object> coreSubscriber) {
        if (this.disposed) {
            if (this.error != null) {
                Operators.error(coreSubscriber, this.error);
                return;
            }
            Operators.complete(coreSubscriber);
            return;
        }
        this.receiver = coreSubscriber;
    }

    protected void onInboundNext(Object msg) {
        if (this.disposed) {
            return;
        }
        if (this.receiver != null) {
            this.receiver.onNext(msg);
            return;
        }
        Queue<Object> q = this.receiverQueue;
        if (q == null) {
            q = new ArrayDeque<>();
            this.receiverQueue = q;
        }
        if (q.offer(msg)) {
            drainReceiver();
            return;
        }
        if (log.isDebugEnabled()) {
            log.debug("Receiver: queue is full");
        }
    }


    @Override
    public void request(long n) {
        if (Operators.validate(n)) {
            this.receiverDemand = Operators.addCap(receiverDemand, n);
            this.drainReceiver();
        }
    }

    @Override
    public void cancel() {
        this.drainReceiver();
    }

    @Override
    public void dispose() {
        this.disposed = true;
        clearQueue(this.receiverQueue);
        this.receiverQueue = null;
    }

    @Override
    public boolean isDisposed() {
        return disposed;
    }

    @Override
    public void subscribe(CoreSubscriber<? super Object> actual) {
        this.startReceiver(actual);
    }
}
